package com.atguigu.flink.worldcount;

import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

public class SocketStreamWordCount {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.socketTextStream("hadoop102" , 10981)
                .flatMap(
                        (String line, Collector<Tuple2<String, Long>> collector) -> {
                                for (String word : line.split(" ")) {
                                    collector.collect(Tuple2.of(word , 1L));
                                }
                            }
                )
                .returns(Types.TUPLE(Types.STRING , Types.LONG))
                .keyBy(data -> data.f0)
                .sum(1)
                .print();

        env.execute();
    }
}

